package cn.heckman.springbatch.batchTask;


import cn.heckman.springbatch.batchTaskCommon.BatchJobListener;
import cn.heckman.springbatch.batchTaskCommon.BatchStepExceptionHandler;
import cn.heckman.springbatch.batchTaskCommon.SpringBatchCommonConfig;
import cn.heckman.springbatch.entity.BlacklistDetectionVo;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepScope;
import org.springframework.batch.item.file.FlatFileItemReader;
import org.springframework.batch.item.file.mapping.DefaultLineMapper;
import org.springframework.batch.item.file.mapping.FieldSetMapper;
import org.springframework.batch.item.file.transform.DelimitedLineTokenizer;
import org.springframework.batch.item.file.transform.FieldSet;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.io.FileSystemResource;
import org.springframework.core.task.TaskExecutor;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.validation.BindException;

import java.io.File;
import java.util.UUID;

/**
 *
 */
@Configuration
@SuppressWarnings("all")
public class LjBatchJob {

    private static final Logger logger = LoggerFactory.getLogger(LjBatchJob.class);

    @Autowired
    public JobBuilderFactory ljJobBuilderFactory;

    @Autowired
    public StepBuilderFactory ljStepBuilderFactory;

    @Autowired
    public PlatformTransactionManager ljPlatformTransactionManager;

    @Autowired
    public BatchStepExceptionHandler ljExceptionHandler;

    @Autowired
    public LjBatchItemProcessor ljBatchItemProcessor;

    @Autowired
    public LjItemWriter ljItemWriter;

    @Autowired
    private TaskExecutor taskExecutor;

    /**
     * 定义job
     *
     * @param listener
     * @return
     */
    @Bean("ljblacklistdetectionjob")
    public Job LjBlacklistDetectionJob(BatchJobListener listener) {
        String uuid = UUID.randomUUID().toString().replaceAll("-", "");
        return ljJobBuilderFactory
                .get("LjBlacklistDetectionJob" + uuid)
                .listener(listener)
                .flow(LjInstanceStep())
                .end()
                .build();
    }

    /**
     * 定义实例step
     *
     * @return
     */
    @Bean
    public Step LjInstanceStep() {
        logger.info("MessageBatchInsertStep");
        return ljStepBuilderFactory.get("MessageBatchInsertStep")
                .<BlacklistDetectionVo, BlacklistDetectionVo>chunk(1000)//指定一次性数据提交的记录数
                .reader(ljItemReader())//读操作
                .processor(ljBatchItemProcessor)//处理操作
                .writer(ljItemWriter)//写操作
                .faultTolerant()//容错
                .skip(Exception.class)//抛出Exception异常时就跳过该条记录的处理
                .skipLimit(1000)//表示当跳过的次数超过数值时则会导致整个Step失败，从而停止继续运行
//                .taskExecutor(new SimpleAsyncTaskExecutor())
                .taskExecutor(taskExecutor)//多线程执行器
                .startLimit(1)//重启的次数
                .allowStartIfComplete(true)//为true告知框架每次重启该Step都要执行
                .exceptionHandler(ljExceptionHandler) // 设置并发方式执行exceptionHandler,异常时打印日志并抛出异常
                .throttleLimit(50) // 并发任务数为 50
                .transactionManager(ljPlatformTransactionManager)//使用默认的 PlatformTransactionManager 对事物进行管理。当配置好事物之后Spring Batch会自动对事物进行管理，无需开发人员显示操作
                .build();
    }


    /**
     * 定义读操作
     *
     * @return
     */
    @Bean
    @StepScope
    public FlatFileItemReader<BlacklistDetectionVo> ljItemReader() {
        FlatFileItemReader<BlacklistDetectionVo> fileRead = new FlatFileItemReader<>();
        fileRead.setEncoding("UTF-8");
        //resources文件夹下有示例数据txt文件
        fileRead.setResource(new FileSystemResource(new File("E:\\gw\\out\\test.txt")));
        //fileRead.setLinesToSkip(2);跳过开头多少行
        DefaultLineMapper<BlacklistDetectionVo> lineMapper = new DefaultLineMapper<BlacklistDetectionVo>();
        lineMapper.setLineTokenizer(new DelimitedLineTokenizer("\r\n"));
        lineMapper.setFieldSetMapper(new FieldSetMapper<BlacklistDetectionVo>() {
            @Override
            public BlacklistDetectionVo mapFieldSet(FieldSet fieldSet) throws BindException {
                BlacklistDetectionVo user = new BlacklistDetectionVo();
                try {
                    String[] strs = fieldSet.readString(0).split(" ");
                    SpringBatchCommonConfig.totalNum.getAndIncrement();
                    user.setPhone(strs[0]);
                    user.setFlag(strs[1]);
                } catch (Exception e) {
                    logger.error("解析异常：" + e.getMessage());
                }
                return user;
            }
        });
        fileRead.setLineMapper(lineMapper);
        return fileRead;
    }

}
